package ApplicationTest.Example.KafKa

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, TaskContext}

object SimpleCreateDirecty {

  private val conf = new SparkConf().setMaster("local[*]").setAppName("Scala Spark Test Application")
  var streamingContext : StreamingContext = new StreamingContext(conf, Seconds(5))

  def main(args: Array[String]): Unit = {


    val kafkaParams = Map[String, Object](
      "metadata.broker.list" -> "master:9092",
      "bootstrap.servers" -> "master:9092,spark02:9092,spark03:9092", //连接两台kafka 服务 -- 关于zookeeper 发现
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "group1", //分配一个组
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("helloword").toSet //本地集群决定 topic

    /**
      * 关于链接 kafka -- 关于消费者consumer的记录
      */
    val lines = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //关于序列化的过程部分

    /** ***************************************************************************************************************
      * 保存偏移度部分
      * （如果在计算的时候失败了，会接着上一次偏移度进行重算，不保存新的偏移度）
      * 计算成功后保存偏移度
      */
    lines.foreachRDD {
      rdd =>
        val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd.foreachPartition {
          iter =>
            val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
            println(s"[ DealFlowBills2 ] --------------------------------------------------------------------")
            println(s"[ DealFlowBills2 ]  topic: ${o.topic}")
            println(s"[ DealFlowBills2 ]  partition: ${o.partition} ")
            println(s"[ DealFlowBills2 ]  fromOffset 开始偏移量: ${o.fromOffset} ")
            println(s"[ DealFlowBills2 ]  untilOffset 结束偏移量: ${o.untilOffset} 需要保存的偏移量,供下次读取使用★★★")
            println(s"[ DealFlowBills2 ] --------------------------------------------------------------------")
            // 写zookeeper
            //zk.offsetWork(s"${o.topic}offset", s"${o.topic},${o.partition},${o.untilOffset}")

        }
    }



    streamingContext.start()  //设置开始
    streamingContext.awaitTermination() //等待执行

  }
}
